package com.dkcrow.hadoop;

import java.net.URI;

import com.dkcrow.hadoop.conf.MyConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.lib.InputSampler;
import org.apache.hadoop.mapred.lib.TotalOrderPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class SequenceFileTotalSort extends Configured implements Tool {

	public int run(String[] args) throws Exception {
		JobConf conf = new JobConf(getConf(), SequenceFileTotalSort.class);
		conf.setJobName("SequenceFileTotalSort");

		conf.setInputFormat(SequenceFileInputFormat.class);
		conf.setOutputFormat(SequenceFileOutputFormat.class);
		conf.setOutputKeyClass(IntWritable.class);
		conf.setPartitionerClass(TotalOrderPartitioner.class);

		// SequenceFile 압축 포맷 설정
		SequenceFileOutputFormat.setCompressOutput(conf, true);
		SequenceFileOutputFormat
				.setOutputCompressorClass(conf, GzipCodec.class);
		SequenceFileOutputFormat.setOutputCompressionType(conf,
				CompressionType.BLOCK);

		// 입출력 경로 설정
		FileInputFormat.setInputPaths(conf, new Path(args[0]));
		FileOutputFormat.setOutputPath(conf, new Path(args[1]));

		Path inputDir = FileInputFormat.getInputPaths(conf)[0];
		inputDir = inputDir.makeQualified(inputDir.getFileSystem(conf));
		Path partitionFile = new Path(inputDir, "_partitions");
		TotalOrderPartitioner.setPartitionFile(conf, partitionFile);

		// 샘플 데이터 추출
		InputSampler.Sampler<IntWritable, Text> sampler = new InputSampler.RandomSampler<IntWritable, Text>(
				0.1, 1000, 10);
		InputSampler.writePartitionFile(conf, sampler);

		URI partitionUri = new URI(partitionFile.toString() + "#_partitions");
		DistributedCache.addCacheFile(partitionUri, conf);
		DistributedCache.createSymlink(conf);

		JobClient.runJob(conf);

		return 0;
	}

    public static final String input = "/user/dkcrow/seq_file/2007";
    public static final String output = "/user/dkcrow/tot_file/2007";

    public static void main(String[] args) throws Exception {
        args = new String[2];
        args[0] = input;
        args[1] = output;
        Configuration conf = MyConfiguration.getConf();
        HdfsHandler.delete(conf, output);
		int res = ToolRunner.run(conf,
				new SequenceFileTotalSort(), args);
		System.out.println("## RESULT:" + res);
	}
}
